package com.loongtech.bi.manager.flow;

import com.loongtech.bi.manager.system.ConfigManager;
import com.loongtech.bi.support.Log;
import com.loongtech.core.util.Time;
import org.apache.hive.jdbc.HiveConnection;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 访问原始数据的manager 
 * @author Marin
 * 
 */
@Service
public class HadoopQueryManager {
	public static final BlockingQueue<String> VISIT_QUEUE = new ArrayBlockingQueue<String>(1000);  // 阻塞队列，最大数量为10。当要执行Hadoop查询时，先执行offer操作，返回true才执行executeSQL；executeSQL执行完毕后，执行poll操作

	@Autowired
	private ConfigManager configManager;
	
	/**
	 * 执行sql语句并返回查询结果
	 * 
	 * @param sql
	 * @return
	 * @throws SQLException
	 */
	public Map.Entry<List<String>, List<Object[]>> executeSQL(String sql) {
		if (null == sql || sql.isEmpty()) {
			Log.bi.error("SQLQueryManager executeSQL sql is empty !");
			VISIT_QUEUE.poll();
			return null;
		}
		
		int randomKey = (int)(Math.random() * 10000); // 随机KEY。用与标识不用sql
		
		System.out.println("begin to execute impala sql，randomKey：" + randomKey + " ，time: " + Time.getNowDateDetailStr());
		
		HiveConnection impalaConnection = null;
		ResultSet rs = null;
		try {
			Class.forName(ConfigManager.impala_jdbc_driver);
			impalaConnection = (HiveConnection)DriverManager.getConnection(configManager.getConfigValueByName(ConfigManager.impala_jdbc_url));
			Statement impalaStatement = impalaConnection.createStatement();
			impalaStatement.execute(configManager.getConfigValueByName(ConfigManager.user_hive_database));
			rs = impalaStatement.executeQuery(sql);
			ResultSetMetaData rsmd = rs.getMetaData();
			int columnNum = rsmd.getColumnCount();
			List<String> nameList = new ArrayList<String>();
			for (int i = 1; i <= columnNum; ++i) {
				nameList.add(rsmd.getColumnName(i));
			}
			List<Object[]> dataList = new LinkedList<Object[]>();
			while (rs.next()) {
				Object[] lineData = new Object[columnNum];
				for (int index = 0; index < columnNum; ++index) {
					lineData[index] = rs.getString(index + 1);
				}
				dataList.add(lineData);
			}
			return new java.util.AbstractMap.SimpleEntry<List<String>, List<Object[]>>(nameList, dataList);
		} catch (Exception e) {
			e.printStackTrace();
			Log.bi.error("SQLQueryManager executeSQL Exception {},{}", sql, e);
			throw new RuntimeException(e);
		} finally {
			String t = VISIT_QUEUE.poll();
			System.out.println("VISIT_QUEUE poll : " + t);
			
			try {
				if (rs != null) {
					rs.close();
				}
				if (impalaConnection != null && !impalaConnection.isClosed()) {
					impalaConnection.close();
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
			
			System.out.println("end to execute impala sql ，randomKey：" + randomKey + " ，time: " + Time.getNowDateDetailStr());
		}
	}
}
